package com.tensai.study.java.newconcept.typereactiveprogram.reactor.schedule;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

/**
 * @author Tensai
 */
@Slf4j
public class ScheduleTest {

	//publishOn和subscribeOn主要用来进行切换Scheduler的执行上下文。
	//先讲一个结论，就是在链式调用中，publishOn可以切换Scheduler，但是subscribeOn并不会起作用。
	//这是因为真正的publish-subscribe关系只有在subscriber开始subscribe的时候才建立。

	/**
	 * publishOn可以在链式调用的过程中，进行publish的切换：
	 *
	 * @throws InterruptedException Interrupted Exception
	 */
	@Test
	public void publishOnTest() throws InterruptedException {
		Scheduler s = Schedulers.newParallel("parallel-scheduler", 4);
		final Flux<String> flux = Flux
				.range(1, 2)
				.map(i -> 10 + i + ":" + Thread.currentThread())
				.publishOn(s)
				.map(i -> "value " + i + ":" + Thread.currentThread());

		new Thread(() -> flux.subscribe(log::info), "ThreadA").start();
		log.info(String.valueOf(Thread.currentThread()));
		Thread.sleep(5000);
	}

	/**
	 * subscribeOn是用来切换Subscriber的执行上下文，不管subscribeOn出现在调用链的哪个部分，
	 * 最终都会应用到整个调用链上。
	 * <br />
	 * <br />
	 * 可以看到,主线程的名字是Thread。Subscriber线程的名字是ThreadA。
	 * <br />
	 * <br />
	 * 那么在publishOn之前，map使用的线程就是ThreadA。
	 * 而在publishOn之后，map使用的线程就切换到了parallel-scheduler线程池。
	 *
	 * @throws InterruptedException Interrupted Exception
	 */
	@Test
	public void subscribeOnTest() throws InterruptedException {
		Scheduler s = Schedulers.newParallel("parallel-scheduler", 4);
		final Flux<String> flux = Flux
				.range(1, 2)
				.map(i -> 10 + i + ":" + Thread.currentThread())
				.subscribeOn(s)
				.map(i -> "value " + i + ":" + Thread.currentThread());

		new Thread(() -> flux.subscribe(log::info), "ThreadA").start();
		log.info(String.valueOf(Thread.currentThread()));
		Thread.sleep(5000);
	}

}
